package cn.xinfei.xdecision.engine.runner.metadata;

import cn.xinfei.xdecision.common.model.enginex.po.Engine;
import cn.xinfei.xdecision.common.model.enginex.po.EngineNode;
import cn.xinfei.xdecision.common.model.enginex.po.EngineVersion;
import cn.xinfei.xdecision.common.service.enginex.EngineNodeService;
import cn.xinfei.xdecision.common.service.enginex.EngineVersionService;
import cn.xinfei.xdecision.engine.runner.executor.NodeHandler;
import cn.xinfei.xdecision.thread.CustomerThreadPoolExecutor;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Table;
import com.google.common.collect.Tables;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

@Slf4j
@Service
public class DecisionFlowLoader {


    @Resource
    private EngineNodeService engineNodeService;

    @Resource
    private EngineVersionService engineVersionService;

    @Resource(name = "metadataCacheReloadThreadPool")
    public CustomerThreadPoolExecutor metadataCacheReloadThreadPool;


    public Map<String, EngineVersion> init(Table<String, Long, Engine> engineAndVersionMap) {
        log.info("DecisionFlowLoader EngineVersion start");
        ConcurrentHashMap<String, EngineVersion> engineVersionMap = new ConcurrentHashMap<>();
        Set<Table.Cell<String, Long, Engine>> cells = engineAndVersionMap.cellSet();
        Iterator<Table.Cell<String, Long, Engine>> iterator = cells.iterator();
        CountDownLatch countDownLatch = new CountDownLatch(engineAndVersionMap.size());
        while (iterator.hasNext()) {
            Table.Cell<String, Long, Engine> next = iterator.next();
            metadataCacheReloadThreadPool.submit(() -> {
                String engineCode = next.getRowKey();
                Engine engine = next.getValue();
                try {
                    //理论上只有一条运行中的版本
                    EngineVersion runningVersion = engineVersionService.getRunningVersion(engineCode);
                    if (null != runningVersion) {
                        runningVersion.setEngineName(engine.getName());
                        engineVersionMap.put(engineCode, runningVersion);
                    } else {
                        log.warn("引擎不存在运行中的版本,engineCode={}", engineCode);
                    }
                } catch (Exception e) {
                    log.info("reload engine config engineCode={}, error:{}", engineCode, e.getMessage(), e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        log.info("DecisionFlowLoader EngineVersion end.size={}", engineVersionMap.size());
        return engineVersionMap;
    }

    public Table<String, String, NodeHandler> init(Map<String, EngineVersion> engineVersionMap) {
        log.info("DecisionFlowLoader NodeHandler start");
        log.info("engineVersionMap====>{}", JSON.toJSONString(engineVersionMap));
        long start = System.currentTimeMillis();
        Table<String, String, NodeHandler> engineAndNodeMap = Tables.synchronizedTable(HashBasedTable.create());
        Set<Map.Entry<String, EngineVersion>> entries = engineVersionMap.entrySet();
        Iterator<Map.Entry<String, EngineVersion>> iterator = entries.iterator();
        int size = engineVersionMap.size();
        CountDownLatch countDownLatch = new CountDownLatch(size);
        log.info("size====>{}", size);
        while (iterator.hasNext()) {
            Map.Entry<String, EngineVersion> next = iterator.next();
            String engineCode = next.getKey();
            EngineVersion engineVersion = next.getValue();
            metadataCacheReloadThreadPool.submit(() -> {
                try {
                    //理论上只有一条运行中的版本
                    Long versionId = engineVersion.getId();

                    List<EngineNode> engineNodeList = engineNodeService.getEngineNodeListByEngineVersionId(versionId);
                    if (CollectionUtils.isEmpty(engineNodeList)) {
                        log.debug("此版本下的引擎，没有可执行节点。engineCode={},version={}", engineCode, versionId);
                        return;
                    }
                    log.info("engineNodeList====>{},versionId={}", engineNodeList.size(), versionId);
                    //查询对应的versionNo
                    engineNodeList.forEach(engineNode -> engineNode.setEngineVersion(
                            engineVersionService.getRunningVersion(engineCode).getVersionNo() + ""));
                    for (int i = 0; i < engineNodeList.size(); i++) {
                        EngineNode engineNode = engineNodeList.get(i);
                        try {
                            engineNode.setEngineCode(engineCode);
                            String nodeType = engineNode.getNodeType();
                            NodeHandler nodeHandler = new NodeHandler(engineNode, nodeType);
                            nodeHandler.init();
                            nodeHandler.setEngineCode(engineCode);
                            engineAndNodeMap.put(engineCode, engineNode.getNodeCode(), nodeHandler);
                        } catch (Exception e) {
                            log.error("获取节点实例异常,engineNode={}", JSON.toJSONString(engineNode), e);
                        }
                    }
                } catch (Exception e) {
                    log.info("reload engine config engineCode={}, error:{}", engineCode, e.getMessage(), e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        long end = System.currentTimeMillis();
        log.info("DecisionFlowLoader NodeHandler end.size={},duration={}", engineAndNodeMap.size(), end - start);
        return engineAndNodeMap;
    }


    public Map<String, EngineVersion> getEngineVersionInfo(Table<String, Long, Engine> engineAndVersionMap) {
        log.info("DecisionFlowLoader EngineVersion start");
        long start = System.currentTimeMillis();
        ConcurrentHashMap<String, EngineVersion> engineVersionMap = new ConcurrentHashMap<>();
        Set<Table.Cell<String, Long, Engine>> cells = engineAndVersionMap.cellSet();
        Iterator<Table.Cell<String, Long, Engine>> iterator = cells.iterator();
        CountDownLatch countDownLatch = new CountDownLatch(engineAndVersionMap.size());
        while (iterator.hasNext()) {
            Table.Cell<String, Long, Engine> next = iterator.next();
            metadataCacheReloadThreadPool.submit(() -> {
                Long versionNo = next.getColumnKey();
                String engineCode = next.getRowKey();
                Engine engine = next.getValue();
                try {
                    List<EngineVersion> runningVersion = engineVersionService.getRunningVersionList(engineCode, versionNo);
                    if (null != runningVersion && runningVersion.size() > 0) {
                        for (EngineVersion version : runningVersion) {
                            version.setEngineName(engine.getName());
                            engineVersionMap.put(engineCode, version);
                        }
                    }
                } catch (Exception e) {
                    log.info("reload engine config engineCode={}, error:{}", engineCode, e.getMessage(), e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        long end = System.currentTimeMillis();
        log.info("DecisionFlowLoader EngineVersion end.size={},duration={}", engineVersionMap.size(), end - start);
        return engineVersionMap;
    }
}
